package com.ak.game.service;

import com.ak.game.util.AjaxResult;
import com.ak.game.util.Md5SignatureUtils;
import com.alibaba.fastjson2.JSONObject;
import com.baomidou.mybatisplus.core.toolkit.ObjectUtils;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.http.DefaultHttpHeaders;
import io.netty.handler.codec.http.HttpClientCodec;
import io.netty.handler.codec.http.HttpHeaders;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.websocketx.*;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslContextBuilder;
import io.netty.handler.ssl.util.InsecureTrustManagerFactory;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.net.URI;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.*;

/**
 * 远程WebSocket客户端服务实现类，用于与WebSocket服务器建立连接并发送/接收消息
 */
@Service
@Slf4j
public class RemoteWebSocketClientService {

    /**
     * 游戏业务Service层
     */
    @Autowired
    private BettingService bettingService;

    /**
     * 配置的远程websocket地址
     */
    @Value("${netty.websocket.uri}")
    private String webSocketUri;

    /**
     * 重试次数
     */
    @Value("${netty.websocket.retry.attempts}")
    private int retryAttempts;

    /**
     * 接口api服务端的验签加密私钥
     */
    @Value("${service-sign}")
    private String serviceSign;

    /**
     * 游戏id
     */
    @Value("${game.id}")
    private String gameId;

    /**
     * Netty的工作线程组，负责处理网络事件
     */
    private EventLoopGroup workerGroup;

    /**
     * 当前与服务器的连接通道
     */
    private SocketChannel channel;

    /**
     * 重连定时器服务
     */
    private ScheduledExecutorService reconnectExecutor;

    /**
     * WebSocket握手处理器
     */
    private WebSocketClientHandshaker handshaker;

    public final static Map<String, CompletableFuture<Object>> FUTURES = new ConcurrentHashMap<>();

    /**
     * 初始化方法，启动WebSocket客户端并设置定时重连
     */
    @PostConstruct
    public void startClient() {
        // 初始化工作线程组
        workerGroup = new NioEventLoopGroup();
        // 初始化单线程定时执行器
        reconnectExecutor = Executors.newSingleThreadScheduledExecutor();
        // 首次尝试连接
        connect();
        // 定时任务，每5秒检查连接状态，如不活动则尝试重新连接
        reconnectExecutor.scheduleWithFixedDelay(this::connect, 0, retryAttempts, TimeUnit.SECONDS);
    }

    /**
     * 实现连接逻辑，如果当前没有活跃连接，则尝试建立新的连接
     */
    private void connect() {
        if (channel != null && channel.isActive()) {
            // 如果已有活跃连接，则直接返回。
            return;
        }
        // 准备Bootstrap以配置客户端
        Bootstrap b = new Bootstrap();
        b.group(workerGroup)
                // 指定使用NIO传输
                .channel(NioSocketChannel.class).handler(new ChannelInitializer<SocketChannel>() {

                    @Override
                    protected void initChannel(SocketChannel ch) {
                        ChannelPipeline p = ch.pipeline();
                        // 如果是HTTPS（wss），则添加SSL处理器。
                        if (webSocketUri.startsWith("wss")) {
                            try {
                                SslContext sslCtx = SslContextBuilder.forClient()
                                        // 测试的信任管理器 非安全的
                                        .trustManager(InsecureTrustManagerFactory.INSTANCE).build();
                                p.addLast(sslCtx.newHandler(ch.alloc()));
                            } catch (Exception e) {
                                throw new RuntimeException(e);
                            }
                        }
                        // 添加HTTP编解码器、消息聚合器以及WebSocket客户端处理器
                        p.addLast(new HttpClientCodec(), new HttpObjectAggregator(65536), new WebSocketClientHandler(handshaker));
                    }
                });

        try {
            // 设置URI请求地址
            URI uri = new URI(webSocketUri);
            // 添加自定义请求头
            HttpHeaders headers = new DefaultHttpHeaders();
            long timestamp = System.currentTimeMillis();
            headers.add("sign", Md5SignatureUtils.generateMd5Sign(String.format("%s%s%d", serviceSign, gameId, timestamp)));
            headers.add("game_id", gameId);
            headers.add("timestamp", String.valueOf(timestamp));
            handshaker = WebSocketClientHandshakerFactory.newHandshaker(uri, WebSocketVersion.V13, null, false, headers);
            // 连接到WebSocket服务器
            ChannelFuture f = b.connect(uri.getHost(), uri.getPort()).sync();
            channel = (SocketChannel) f.channel();
            log.info("连接到 {} 成功", webSocketUri);
            // 添加WebSocket协议处理器到管道
            channel.pipeline().addLast(new WebSocketClientProtocolHandler(handshaker));
            // 监听连接完成事件，成功后进行握手
            f.addListener((ChannelFutureListener) future -> {
                if (future.isSuccess()) {
                    handshaker.handshake(channel);
                } else {
                    future.cause().printStackTrace();
                    // 可重试....
                }
            });
        } catch (Exception e) {
            e.printStackTrace();
            log.error("连接到 {} 失败", webSocketUri);
        }
    }

    /**
     * 发送消息的方法
     *
     * @param ajaxResult Ajax消息体
     */
    public void sendMessageAjax(AjaxResult ajaxResult) {
        if (channel != null && channel.isActive()) {
            // 通过WebSocketClientHandler内部类来发送消息
            channel.pipeline().get(WebSocketClientHandler.class).sendWebSocketMessageAjax(channel.pipeline().context(WebSocketClientHandler.class), ajaxResult);
        } else {
            log.error("连接已关闭，发送消息失败={}", ajaxResult);
        }
    }

    /**
     * 发送消息的方法
     *
     * @param message 消息体
     */
    public CompletableFuture<Object> sendMessage(String message) {
        CompletableFuture<Object> future = new CompletableFuture<>();
        JSONObject jsonObject = JSONObject.parseObject(message);
        // 将请求标识符和CompletableFuture关联起来
        if (ObjectUtils.isNotNull(jsonObject.get("data"))) {
            HashMap dataMap = (HashMap) jsonObject.get("data");
            String userId = dataMap.get("user_id").toString();
            FUTURES.put(userId, future);
        } else {
            HashMap gameMap = (HashMap) jsonObject.get("game");
            String roundId = gameMap.get("round_id").toString();
            FUTURES.put(roundId, future);
        }
        if (channel != null && channel.isActive()) {
            // 在消息中附加请求标识符
            channel.pipeline().get(WebSocketClientHandler.class).sendWebSocketMessage(channel.pipeline().context(WebSocketClientHandler.class), message);
        } else {
            future.completeExceptionally(new IllegalStateException("WebSocket connection is not active."));
        }
        return future;


//        if (channel != null && channel.isActive()) {
//            // 通过WebSocketClientHandler内部类来发送消息
//            channel.pipeline().get(WebSocketClientHandler.class).sendWebSocketMessage(channel.pipeline().context(WebSocketClientHandler.class), message);
//        } else {
//            log.error("连接已关闭，发送消息失败={}", message);
//        }
    }

    /**
     * 内部类 处理WebSocket客户端的网络事件
     */
    class WebSocketClientHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
        private final WebSocketClientHandshaker handshaker;
        private ChannelPromise handshakeFuture;

        public WebSocketClientHandler(WebSocketClientHandshaker handshaker) {
            this.handshaker = handshaker;
        }

        @Override
        public void handlerAdded(ChannelHandlerContext ctx) {
            // 初始化握手成功的Future
            handshakeFuture = ctx.newPromise();
        }

        @Override
        public void channelActive(ChannelHandlerContext ctx) {
            // WebSocketClientProtocolHandler会自动处理握手，此处无需额外操作
        }

        /**
         * 异常处理，关闭通道资源，防止资源泄露
         *
         * @param ctx   通道代理上下文
         * @param cause 异常
         */
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
            cause.printStackTrace();
            // 异常捕获，关闭通道。
            if (channel != null && channel.isActive()) {
                channel.close();
            }
        }

        /**
         * 实现发送WebSocket消息的方法
         *
         * @param ctx        ChannelHandlerContext 上下文对象
         * @param ajaxResult 要发送的Ajax对象
         */
        public void sendWebSocketMessageAjax(ChannelHandlerContext ctx, AjaxResult ajaxResult) {
            ctx.channel().writeAndFlush(new TextWebSocketFrame(JSONObject.toJSONString(ajaxResult)));
        }

        /**
         * 实现发送WebSocket消息的方法
         *
         * @param ctx     ChannelHandlerContext 上下文对象
         * @param message 要发送的字符串对象
         */
        public void sendWebSocketMessage(ChannelHandlerContext ctx, String message) {
            ctx.channel().writeAndFlush(new TextWebSocketFrame(message));
        }


        @Override
        protected void channelRead0(ChannelHandlerContext channelHandlerContext, TextWebSocketFrame textWebSocketFrame) throws Exception {
            String response = textWebSocketFrame.text();
            JSONObject jsonObject = JSONObject.parseObject(response);
            HashMap dataMap = (HashMap) jsonObject.get("data");
            String msg = jsonObject.get("msg").toString();
            String id = "";
            switch (dataMap.get("msgId").toString()) {
                // 登录的返回
                case "106":
                    id = dataMap.get("userId").toString();
                    CompletableFuture<Object> future106 = FUTURES.remove(id);
                    Object isLogin = dataMap.get("isLogin");
                    if (future106 != null) {
                        future106.complete(isLogin);
                    }
                    break;
                // 用户下注的返回
                case "100":
                    id = dataMap.get("userId").toString();
                    Integer code100 = jsonObject.getInteger("code");
                    CompletableFuture<Object> future100 = FUTURES.remove(id);
                    if (future100 != null) {
                        if (code100 == 200) {
                            future100.complete(AjaxResult.success());
                        } else {
                            future100.complete(AjaxResult.fail(msg));
                        }
                    }
                    break;
                // 所有用户本局比赛的结果的返回
                case "104":
                    id = dataMap.get("roundId").toString();
                    Integer code104 = jsonObject.getInteger("code");
                    CompletableFuture<Object> future104 = FUTURES.remove(id);
                    if (code104 == 200) {
                        future104.complete(true);
                    } else {
                        future104.complete(false);
                    }
                    break;
                // 刷新游戏配置
                case "119":
                    bettingService.flushConfig();
                    break;
                // 刷新游戏杀率
                case "120":
                    bettingService.flushWinRate();
                    break;
            }
        }

        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            super.channelRead(ctx, msg);
        }
    }

    /**
     * 当该bean销毁时调用，用于清理资源，防止资源泄露，内存溢出
     */
    @PreDestroy
    public void stopClient() {
        // 关闭通道
        if (channel != null) {
            channel.close();
        }
        // 关闭工作组线程
        workerGroup.shutdownGracefully();
        // 关闭重连定时器
        reconnectExecutor.shutdown();
    }

}
